<?php declare(strict_types=1);


namespace App\Rpc\Service;
use App\Rpc\Lib\NotifyInterface;
use Swoft\Bean\Annotation\Mapping\Inject;
use Swoft\Breaker\Annotation\Mapping\Breaker;
use Swoft\Rpc\Server\Annotation\Mapping\Service;

/**
 * Class NotifyService
 *
 * @since 2.0
 *
 * @Service()
 */
class NotifyService implements NotifyInterface
{

    /**
     * @Inject("rabbit.pool")
     * @var \Six\Rabbit\Pool
     */
    private $rabbit;

    /**
     * 发布消息
     * @Breaker(fallback="fallback")
     * @param $data
     * @return array
     */
    public function publish($data):array
    {
        sgo(function (){
            try{
                $delayTime = 2; //参数控制
                //缓存队列
                $cacheExchangeName = "cacheExchange";
                $routeKey = "/cache";
                $cacheQueueName = 'cache';

                #延迟队列
                $delayExchangeName = 'delayExchange';
                $delayRouteKey = '/delay';
                $delayQueueName = 'delay';

                $connection = $this->rabbit->connect();

                $connectionRabbit = $connection->connection;
                //创建连接
//            $connection = new \PhpAmqpLib\Connection\AMQPStreamConnection('127.0.0.1','5672','guest','guest');
                $channel = $connectionRabbit->channel(); //建立通道



                //是希望消息的存活周期达到时，把消息转移到死信队列当中
                $table= new \PhpAmqpLib\Wire\AMQPTable();
                $table->set('x-dead-letter-exchange',$delayExchangeName); //消息死亡之后转移到的交换机
                $table->set('x-dead-letter-routing-key',$delayRouteKey); //消息死亡之后绑定的路由key
                $table->set('x-message-ttl',$delayTime);   //设置消息存活时间

                $channel->exchange_declare($cacheExchangeName, \PhpAmqpLib\Exchange\AMQPExchangeType::DIRECT, false, true, false);
                //声明一个队列，发送消息使用:(注意参数位置及如果之前已经生成了队列，不能覆盖修改)
                $channel->queue_declare($cacheQueueName,false,true,false,false,false,$table);
                $channel->queue_bind($cacheQueueName,$cacheExchangeName,$routeKey);



                #死信延迟队列
                $channel->exchange_declare($delayExchangeName, \PhpAmqpLib\Exchange\AMQPExchangeType::DIRECT, false, true, false);
                $channel->queue_declare($delayQueueName, false, true, false, false);
                $channel->queue_bind($delayQueueName,$delayExchangeName,$delayRouteKey);

                $msg_body=[
                    'msg_id'=>uniqid(),
                    'version'=>1,
                    'create_time'=>time(),
                    'message_body'=>['order_id'=>12133,'shop_id'=>2],
                    'notify_url'=>'http://127.0.0.1:9805/shop/notify',
                    'notify'=>[1=>5,2=>10,3=>15],//单位为秒
                    'notify_retries_number'=>0, //重试次数，
                    'default_delay_time'=>1,//毫秒为单位
                    'status'=>1, //消息状态
                ];

                /**
                 * 创建AMQP消息类型
                 * delivery_mode 消息是否持久化
                 * AMQPMessage::DELIVERY_MODE_NON_PERSISTENT  不持久化
                 * AMQPMessage::DELIVERY_MODE_PERSISTENT      持久化
                 */
                $msg = new \PhpAmqpLib\Message\AMQPMessage(json_encode($msg_body),['delivery_mode' => \PhpAmqpLib\Message\AMQPMessage::DELIVERY_MODE_PERSISTENT]);

                #发布消息

                #发布消息
                $result = $channel->basic_publish($msg,$cacheExchangeName,$routeKey);
                if($result){
                    $res =  ['status' => 1 ,'result' => '消息发布成功'] ;
                }else{
                    $res =  ['status' => 1 ,'result' => '消息发布成功'] ;
                }
//            var_dump($result);

                $connection->setRelease(true); //设置重新使用
                $connection->release();
                $channel->close();
                $connection->close();
                return $res;
            }catch (\Exception $exception){
                var_dump($exception->getMessage());
            }
        });
    }

    public function fallback()
    {
        return ['status' => 1 ,'result' => '服务降级了'];
    }
}